package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.factory.DataHubSourceFactory;
import cn.gwm.flink.streaming.function.process.OriginalDataProcessFunction;
import cn.gwm.flink.streaming.ods.ModelHandle;
import cn.gwm.flink.streaming.ods.model.ModelCanEnum;
import cn.gwm.flink.streaming.ods.model.StandardFieldConf;
import cn.gwm.utils.ConfigLoader;
import cn.gwm.utils.DateTimeUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.db.Db;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


public class OtspEs11Task {

    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        // 3.1）设置每隔30s一个周期开启checkpoint（checkpoint开启周期，不能太长也不能太短：1-5分钟）
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        setEnv(env,"OtspEs11Task");
        Map<String, OutputTag<String>> canTagMap = new ConcurrentHashMap<>(16);
        canTagMap.put(ModelCanEnum.ES11.getModelName(), new OutputTag<String>(ModelCanEnum.ES11.getValue()) {});
        SingleOutputStreamOperator<String> process =  env.addSource(DataHubSourceFactory.getDatahubSourceFunction3("otsp_pdcu_fault_detection_data_dev",
                        "1686106493568DOBI5", DateTimeUtil.defaultFormatPare(ConfigLoader.get("all.receive.endTime"))))
                .process(new OriginalDataProcessFunction(canTagMap)).returns(Types.STRING).uid("ods_source_id_otsp");
        CleanData cleanData = new CleanData();
        SingleOutputStreamOperator<JSONObject> operator = process.getSideOutput(canTagMap.get(ModelCanEnum.ES11.getModelName()))
                .flatMap(cleanData).returns(JSONObject.class).name("flatMapEs11Otsp");
//        operator.print();
        operator.addSink(BaseTask.getOrcTxtSink("ods_can_es11","ods/otsp/can", cleanData.getList()))
                .name("to_otsp_CanHive_es11").uid("ods_otsp_toCanHive_id_es11");
//        DataStream<JSONObject> gpsStream = stream.getSideOutput(gpsTag);
//        gpsStream.addSink(BaseTask.getOrcTxtSink("ods_gps_es11","ods/otsp/gps",
//                        BeanSource.SourceEnum.gpsIntranet)).name("to_otsp_GPSHive_es11")
//                .uid("ods_otsp_toGpsHive_id_es11");
        env.execute("OtspEs11Task");
    }

    static class SideOutDataProcessFunction extends ProcessFunction<JSONObject, JSONObject>{

        private OutputTag<JSONObject> gpsTag;

        public SideOutDataProcessFunction(OutputTag<JSONObject> gpsTag){
            this.gpsTag = gpsTag;
        }

        @Override
        public void processElement(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
            if(BaseFields.COMMAND_HW_HEARBEAT_DECODED.equals(object.getStr(BaseFields.command))){
                out.collect(object);
            }else if(BaseFields.COMMAND_HW_LOCATION.equals(object.getStr(BaseFields.command))){
                ctx.output(gpsTag,object);
            }
        }
    }

    private static void setEnv(StreamExecutionEnvironment env,String taskName) {
        env.enableCheckpointing(150 * 1000L);
        // 启动类型
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        // 3.2）设置检查点的mode，exactly-once，保证数据的一次性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 3.3）设置两次checkpoint之间的时间间隔，避免两次间隔时间太近导致频繁的checkpoint
        env.getCheckpointConfig().setCheckpointInterval(150*1000L);
        // 3.4）设置checkpoint的超时时间（一般设置一个checkpoint周期的二分之一）
        env.getCheckpointConfig().setCheckpointTimeout(80 * 1000L);
        // 3.5）设置checkpoint的最大尝试次数，同一个时间内有几个checkpoint在运行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        // 3.6）设置job作业取消的时候，是否保留checkpoint的计算结果
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 3.7）设置job作业运行过程中，如果checkpoint失败，是否停止作业
        env.getCheckpointConfig().setFailOnCheckpointingErrors(true);
        // 3.8）设置checkpoint的存储后端，使用rocksdb作为状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        // 设置可容忍的检查点的失败数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        //启用不对齐检查点保存方式
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        env.getCheckpointConfig().setCheckpointStorage(ConfigLoader.get("hdfsUri") + "/flink/checkpoints/" + taskName);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.minutes(15),Time.seconds(10)));
    }


    static class CleanData implements FlatMapFunction<String, JSONObject> {
        List<String> list;
        Map<String,String> codeMap;
        public CleanData(){
            initList();
            System.out.println("clean 构造函数");
        }
        private void initList() {
            this.list = new ArrayList<>();
            this.codeMap = new ConcurrentHashMap<>(16);
            try {
                Db.use(DSFactory.get(ConfigLoader.get("mysql.group.db.config")))
                        .find(StandardFieldConf.getEntityWhere(), StandardFieldConf.class).stream()
                        .filter(data -> ModelHandle.BIZ_TYPE_ODS.equals(data.getBizType())
                                &&Integer.parseInt(data.getState())>0
                                &&(("1".equals(data.getVersion())
                                &&"ES11".equals(data.getVehicletype()))
                                ||"BASE".equals(data.getVehicletype())))
                        .sorted(new Comparator<StandardFieldConf>() {
                            @Override
                            public int compare(StandardFieldConf o1, StandardFieldConf o2) {
                                return Integer.parseInt(o1.getId())-Integer.parseInt(o2.getId());
                            }
                        })
                        .forEach(data -> {
                            list.add(data.getSemaphore());
                            codeMap.put(data.getSubCode(),data.getSemaphore());
                        });
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        }
        public  List<String>  getList(){
            return this.list;
        }
        @Override
        public void flatMap(String str, Collector<JSONObject> collector) throws Exception {
            JSONObject object = null;
            try {
                object = JSONUtil.parseObj(str);
            } catch (Exception e) {
                e.printStackTrace();
                throw e;
            }
            //周期can信号 ||BaseFields.COMMAND_HW_LOCATION.equals(object.getStr(BaseFields.command))
            if(BaseFields.COMMAND_HW_HEARBEAT_DECODED.equals(object.getStr(BaseFields.command))){
                JSONObject resultOjb = JSONUtil.createObj();
                for (Map.Entry<String,String> entry : codeMap.entrySet()) {
                    if(object.containsKey(entry.getKey())){
                        resultOjb.set(entry.getValue(), object.getStr(entry.getKey()));
                    }
                }
                collectData(collector, resultOjb);
            }
        }
        private void collectData(Collector<JSONObject> collector, JSONObject object) {
            try {
                object.set(BaseFields.STORE_TIME, System.currentTimeMillis())
                        .set(VehicleType.VEHICLE_TYPE, "ES11")
                        .set(BaseFields.PARTITION_KEY, DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
                                DatePattern.NORM_DATE_PATTERN));
            } catch (Exception e) {
                e.printStackTrace();
                throw e;
            }
            object.set(BaseFields.tid,object.getLong(BaseFields.tid)/1000*1000);
            if(object.getLong(BaseFields.tid)>=1688745600000L&&object.getLong(BaseFields.tid)<1689091200000L){
                collector.collect(object);
            }
        }
    }
}
